package com.bw.yk06;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

public class test9 {
    public static void main(String[] args) throws Exception {
        //创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);
        //读取kafka
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop-single:9092");
        properties.setProperty("group.id", "test3");
        DataStreamSource<String> stream = env.addSource(new FlinkKafkaConsumer<>("tms_ods", new SimpleStringSchema(), properties));
        //1.当日发单总数，当日转运完成运单数，当日完成运输次数，当日完成运输里程，当日完成运输时长(10分)
        SingleOutputStreamOperator<Tuple6<String, Integer, Integer, Integer, Double, Long>> reduce = stream
                .filter(x -> JSON.parseObject(x).getString("table").equals("transport_task") &&
                        JSON.parseObject(x).getJSONObject("data") != null &&
                        JSON.parseObject(x).getJSONObject("data").size() > 0
                )
                .map(new MapFunction<String, Tuple6<String, Integer, Integer, Integer, Double, Long>>() {
                    @Override
                    public Tuple6<String, Integer, Integer, Integer, Double, Long> map(String s) throws Exception {
                        JSONObject jo = JSON.parseObject(s).getJSONObject("data");
                        String day = jo.getString("create_time").substring(0, 10);
                        int order_cnt = 0; //单总数
                        int complete_cnt = 0; //完成运单数
                        int trans_cnt = 0; //完成运输次数
                        double trans_dists = 0; //完成运输里程
                        long trans_ts = 0; //完成运输时长

                        order_cnt = jo.getInteger("order_num");
                        if (jo.getString("status").equals("67004")) {
                            complete_cnt = jo.getInteger("order_num");
                            trans_cnt++;
                            trans_dists = jo.getDouble("actual_distance");
                            trans_ts = jo.getLong("actual_end_time") - jo.getLong("actual_start_time");
                        }
                        return new Tuple6<>(day, order_cnt, complete_cnt, trans_cnt, trans_dists, trans_ts);
                    }
                }).keyBy(x -> x.f0)
                .reduce((x, y) -> new Tuple6<>(x.f0, x.f1 + y.f1, x.f2 + y.f2, x.f3 + y.f3, x.f4 + y.f4, x.f5 + y.f5));


        reduce.print();



        env.execute();
    }
}
